package com.navinfo.opentsp.platform.computing.analysis.service;

import com.navinfo.opentsp.platform.computing.analysis.util.PropertiesUtil;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;

import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.URISyntaxException;
import java.util.*;

public class IdleSpeedService {

    private static Map<String, String> config = new HashMap<>();

    public SparkSession getSpark(){
        SparkSession.Builder sparkSessionBuilder = SparkSession.builder();
        Set<Map.Entry<String, String>> entrySet = getConfig().entrySet();
        for(Map.Entry<String, String> entry : entrySet){
            String key = entry.getKey();
            String value = entry.getValue();
            if(!key.startsWith("spark")){
                continue;
            }
            sparkSessionBuilder.config(key, value);
        }
        sparkSessionBuilder.enableHiveSupport();
        return sparkSessionBuilder.getOrCreate();
    }

    public JavaSparkContext createSparkContext(SparkSession spark){
        return new JavaSparkContext(spark.sparkContext());
    }

    public Map<String, String> getConfig(){
        if(config.size() == 0){
            try {
                File directory = new File(PropertiesUtil.class.getProtectionDomain().getCodeSource().getLocation().toURI().getPath()).getParentFile();
                Properties properties = new Properties();
                InputStream inputStream = new FileInputStream(directory.getAbsolutePath() + File.separator + "idle_speed.properties");
                properties.load(inputStream);
                Enumeration<Object> keys = properties.keys();
                while(keys.hasMoreElements()){
                    Object key = keys.nextElement();
                    String value = properties.get(key).toString();
                    config.put(key.toString(), value);
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        return config;
    }

}
